from pyspark.sql import SparkSession
import os 

spark = SparkSession.builder.enableHiveSupport().appName("dim_xl_caft_promotion").getOrCreate()

video_id_txt='/home/etlctl/hadoop/dim/dim_xl_caft_promotion_video.txt'
live_id_txt = '/home/etlctl/hadoop/dim/dim_xl_caft_promotion_live.txt'


spark.sql("drop table if exists temp_caft_promotion_video")
spark.sql("create table if not exists temp_caft_promotion_video (video_id string)")
spark.sql(f"LOAD DATA LOCAL INPATH '{video_id_txt}' INTO TABLE temp_caft_promotion_video")
spark.sql("insert overwrite table dim.dim_xl_caft_promotion_video select '102503' as tenant_id,video_id from temp_caft_promotion_video group by video_id")

spark.sql("drop table if exists temp_caft_promotion_live")
spark.sql("create table if not exists temp_caft_promotion_live (room_id string)")
spark.sql(f"LOAD DATA LOCAL INPATH '{live_id_txt}' INTO TABLE temp_caft_promotion_live")
spark.sql("insert overwrite table dim.dim_xl_caft_promotion_live select '102503' as tenant_id,room_id from temp_caft_promotion_live group by room_id")

spark.stop()
